﻿/*
   Copyright 2013 Dmitry Bratus

   Licensed under the Apache License, Version 2.0 (the "License");
   you may not use this file except in compliance with the License.
   You may obtain a copy of the License at

       http://www.apache.org/licenses/LICENSE-2.0

   Unless required by applicable law or agreed to in writing, software
   distributed under the License is distributed on an "AS IS" BASIS,
   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   See the License for the specific language governing permissions and
   limitations under the License.
*/
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading;
using System.Collections.Concurrent;
using System.Threading.Tasks;

namespace Flower.Actors
{
    public sealed class MultiActor<TProcessor, TMessage> 
        : IActor<TMessage>
        where TMessage : IMessage<TProcessor>
    {
        private const int RebalancingIntervalSec = 1;
        private const int RebalancingMaxDepth = 3;
        private const int RebalancingThreshold = 2;

        private readonly IActor<TMessage>[] _actors;
        private readonly PriorityQueue<TMessage>[] _queueCollections; 
        private readonly Timer _rebalancer;
        
        private long _index;

        public event EventHandler<ActorExceptionEventArgs<TMessage>> Error;

        public IProducerConsumerCollection<TMessage> Queue
        {
            get { return null; }
        }

        public MultiActor(int size, Func<int, IActor<TMessage>> actorsFactory)
        {
            _actors = new IActor<TMessage>[size];
            _queueCollections = new PriorityQueue<TMessage>[size];

            int queuesCount = 0;

            for (int i = 0; i < size; i++)
            {
                _actors[i] = actorsFactory(i);
                _actors[i].Error += OnError;

                _queueCollections[i] = _actors[i].Queue as PriorityQueue<TMessage>;

                if (_queueCollections[i] != null)
                {
                    ++queuesCount;
                }
            }

            _rebalancer =
                queuesCount > 1 ?
                    new Timer
                    (
                        Rebalance,
                        0,
                        TimeSpan.FromSeconds(RebalancingIntervalSec),
                        TimeSpan.FromSeconds(RebalancingIntervalSec)
                    ) : 
                    null;
        }

        private void Rebalance(object arg)
        {
            var depth = (int)arg;
            int maxLen = 0;
            int indexOfMax = -1;
            int minLen = int.MaxValue;

            for (int i = 0; i < _queueCollections.Length; i++)
            {
                PriorityQueue<TMessage> q = _queueCollections[i];

                if (q != null)
                {
                    int len = q.Count;

                    if (len > maxLen)
                    {
                        maxLen = len;
                        indexOfMax = i;
                    }

                    if (len < minLen)
                    {
                        minLen = len;
                    }
                }
            }

            if (maxLen > RebalancingThreshold && (double)(maxLen - minLen) / maxLen > 0.5)
            {
                TMessage[] stolenWork;

                lock (_queueCollections[indexOfMax])
                {
                    stolenWork = _queueCollections[indexOfMax].Steal(0.5);
                }

                int prev = 0;
                int dest = 0;

                foreach (TMessage message in stolenWork)
                {
                    while(dest == prev || dest == indexOfMax || _queueCollections[dest] == null)
                    {
                        dest = (dest + 1) % _queueCollections.Length;
                    }

                    _actors[dest].Send(message);

                    prev = dest;
                }

                if (depth < RebalancingMaxDepth)
                {
                    Rebalance(depth + 1);
                }
            }
        }

        private void OnError(object sender, ActorExceptionEventArgs<TMessage> processorExceptionEventArgs)
        {
            if (Error != null)
            {
                Error.Invoke(sender, processorExceptionEventArgs);
            }
        }

        private IActor<TMessage> GetAny()
        {
            return _actors[Interlocked.Increment(ref _index) % _actors.Length];
        }

        public void Send(TMessage message)
        {
            GetAny().Send(message);
        }

        public void Broadcast(TMessage message)
        {
            foreach (IActor<TMessage> a in _actors)
            {
                a.Send(message);
            }
        }

        public void Broadcast(Func<TMessage> message)
        {
            foreach (IActor<TMessage> a in _actors)
            {
                a.Send(message());
            }
        }

        public void Dispose()
        {
            if (_rebalancer != null)
            {
                _rebalancer.Dispose();
            }

            foreach (IActor<TMessage> a in _actors)
            {
                a.Dispose();
            }
        }

        public void Stop()
        {
            Parallel.ForEach(_actors, a => a.Stop());
        }

        public void Resume()
        {
            foreach (IActor<TMessage> a in _actors)
            {
                a.Resume();
            }
        }

        public bool WaitAllMessages(TimeSpan? timeout)
        {
            var results = new BlockingCollection<bool>();

            Parallel.ForEach
            (
                _actors,
                a =>
                {
                    try
                    {
                        results.Add(a.WaitAllMessages(timeout));
                    }
                    catch (Exception)
                    {
                        results.Add(false);
                    }
                }
            );

            bool result = true;

            for (int i = 0; i < _actors.Length; i++)
            {
                result &= results.Take();
            }

            return result;
        }

        public bool IsSuspended { get { return _actors.All(a => a.IsSuspended); } }
    }
}
